package demo.spark.stream

import java.io.{BufferedReader, BufferedWriter, DataOutputStream, FileOutputStream, FileWriter, InputStreamReader, PrintWriter}
import java.net.{InetSocketAddress, ServerSocket, SocketAddress}
import java.util.{Base64, Date, UUID}
import java.util.concurrent.atomic.AtomicInteger

object StreamTestRunner {

  val MONITOR_DIR_TEXT: String = "/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/local_text/spark-stream/";
  val MONITOR_DIR_BINARY: String = "/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/local_binary/spark-stream/";
  val CHECK_POINT_PATH: String = "/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/check_point/";
  val CHECK_POINT_MONITOR_FILE: String = "/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/monitor_dir/test_file.txt";

  def main(args: Array[String]): Unit = {
    //testTCPServerMonitor(port = 9999);
    //testTextFileChangeMonitor();
    testBinaryFileChangeMonitor(charset = "utf-8");
  }


  def testTCPServerMonitor(port: Int = 9999): Unit = {
    require(port != null)
    // 使用JAVA Socket创建一个TCP服务端，使用SparkStream去监听.
    val socketServer: ServerSocket = new ServerSocket();
    socketServer.bind(new InetSocketAddress("localhost",port))
    // 阻塞等待SparkStream客户端连接
    val socket = socketServer.accept()
    val bufferWrite: PrintWriter = new PrintWriter(socket.getOutputStream,true);
    val consoleReader: BufferedReader = new BufferedReader(new InputStreamReader(System.in));

    try {
      // 不关注来自客户端的数据，由服务端向Channel中单向输出, 提供给SparkStream做数据分析的数据源.
      var line: String = null;
      while ((line = consoleReader.readLine()) != null){
        bufferWrite.println(line)
      }
    }finally {
      socket.shutdownInput();
      bufferWrite.close();
      socket.close();
      socketServer.close();
    }
  }


  /**
   * 这里使用本地文件系统进行【文本】测试，HDFS需要连接HDFS文件系统，采用Append的方式记性追加.
   * @param monitorDir
   */
  def testTextFileChangeMonitor(monitorDir: String = MONITOR_DIR_TEXT): Unit = {
    require(monitorDir != null)
    val dir = new java.io.File(monitorDir)
    if(!dir.exists() || dir.isFile){
      println("#### test dir required a Directory for sub files generic, application will exist.")
      System.exit(1);
    }
    var dirPath = monitorDir;
    if(!dirPath.endsWith("/")){
      dirPath = dirPath concat "/"
    }

    val count: AtomicInteger = new AtomicInteger(0)
    val enhancerWriteToMap: BufferedWriter => (Integer, BufferedWriter) = writer => (count.getAndIncrement(),writer);

    // 在给定的目录下(local) 创建6个文件，秒数 mod / 10 为本次输入文本落地的文件！
    val genericRandomFile: () => String = () => dirPath concat UUID.randomUUID().toString.replace("-","") concat ".txt";
    val fileWriterMapped: Map[Integer,BufferedWriter] = Stream.iterate(genericRandomFile.apply())(_ => genericRandomFile.apply())
      .take(6)
      .map(new java.io.File(_))
      .map(file => new BufferedWriter(new FileWriter(file,true)))
      .map(enhancerWriteToMap.apply(_))
      .toMap[Integer,java.io.BufferedWriter];

    // 读取来自console的输入，每行为一个读取周期，写入到 time : mod / 10 的文件中.
    val consoleReader: BufferedReader = new BufferedReader(new InputStreamReader(System.in));
    var line: String = null;
    try {
      while ((line = consoleReader.readLine()) != null && !line.equals("bye")){
        val writer: Option[BufferedWriter] = fileWriterMapped.get(new Date().getSeconds / 10)
        writer.get.write(line)
        writer.get.flush()
      }
    }finally {
      for(entry <- fileWriterMapped.toList){
        entry._2.close();
      }
    }
  }


  /**
   * 这里使用本地文件系统进行【二进制】测试，HDFS需要连接HDFS文件系统，采用Append的方式记性追加.
   * @param monitorDir
   */
  def testBinaryFileChangeMonitor(monitorDir: String = MONITOR_DIR_BINARY, charset: String = "ISO-8859-1"): Unit = {
    require(monitorDir != null)
    val dir = new java.io.File(monitorDir)
    if(!dir.exists() || dir.isFile){
      println("#### test dir required a Directory for sub files generic, application will exist.")
      System.exit(1);
    }
    var dirPath = monitorDir;
    if(!dirPath.endsWith("/")){
      dirPath = dirPath concat "/"
    }

    val count: AtomicInteger = new AtomicInteger(0)
    val enhancerWriteToMap: DataOutputStream => (Integer, DataOutputStream) = outStream => (count.getAndIncrement(),outStream);

    // 在给定的目录下(local) 创建6个文件，秒数 mod / 10 为本次输入文本落地的文件！
    val genericRandomFile: () => String = () => dirPath concat UUID.randomUUID().toString.replace("-","")
    val fileWriterMapped: Map[Integer,DataOutputStream] = Stream.iterate(genericRandomFile.apply())(_ => genericRandomFile.apply())
      .take(6)
      .map(new java.io.File(_))
      .map(file => new DataOutputStream(new FileOutputStream(file,true)))
      .map(enhancerWriteToMap.apply(_))
      .toMap[Integer,java.io.DataOutputStream];

    // 读取来自console的输入，每行为一个读取周期，写入到 time : mod / 10 的文件中.
    val consoleReader: BufferedReader = new BufferedReader(new InputStreamReader(System.in));
    var line: String = null;
    try {
      while ((line = consoleReader.readLine()) != null && !line.equals("bye")){
        val outStream: Option[DataOutputStream] = fileWriterMapped.get(new Date().getSeconds / 10)
        //outStream.get.write(Base64.getEncoder.encode(line.getBytes("utf-8")))
        outStream.get.write(line.getBytes(charset))
        /*val chrAry: Array[Char] = line.toCharArray
        val splitBuffer: StringBuilder = new StringBuilder
        val length = chrAry.length
        for (idx <- 0 until length){
          splitBuffer append Integer.toBinaryString(chrAry(idx))
        }
        outStream.get.writeChars(splitBuffer.toString())*/
        outStream.get.flush()
      }
    }finally {
      for(entry <- fileWriterMapped.toList){
        entry._2.close();
      }
    }
  }


}
